package net.xuele.learn.mq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

@Slf4j
@Data
@Configuration
public class RocketMQConsumerConfiguration {

    /**
     * The Namesrv addr.
     */
    @Value("${rocketmq.consumer.namesrvAddr}")
    private String namesrvAddr;
    /**
     * The Group name.
     */
    @Value("${rocketmq.consumer.groupName}")
    private String groupName;

    /**
     * The Consume thread min.
     */
    @Value("${rocketmq.consumer.consumeThreadMin}")
    private int consumeThreadMin;
    /**
     * The Consume thread max.
     */
    @Value("${rocketmq.consumer.consumeThreadMax}")
    private int consumeThreadMax;

    @Bean
    public MessageProcessor messageProcessor() {
        return new MessageProcessor();
    }

    @Bean
    public MessageListener messageListener(MessageProcessor messageProcessor) {
        MessageListener messageListener = new MessageListener();
        messageListener.setMessageProcessor(messageProcessor);
        return messageListener;
    }
    @Bean(initMethod = "start", destroyMethod = "shutdown")
    @Primary
    public DefaultMQPushConsumer defaultMQPushConsumer(MessageListener messageListener) throws RocketMQException {
        if (StringUtils.isBlank(groupName)) {
            throw new RocketMQException("consumerGroup is null !!!");
        }
        if (StringUtils.isBlank(namesrvAddr)) {
            throw new RocketMQException("namesrvAddr is null !!!");
        }

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(namesrvAddr);


        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.registerMessageListener(messageListener);
        try {
            // +处理缓存阿里数据用于延迟发送
            consumer.subscribe(Constant.TOPIC, Constant.CLAIM_TAG + "||" + Constant.ACK_TAG);
            log.info("consumer is start !!! consumerGroup:{},namesrvAddr:{}", groupName, namesrvAddr);
        } catch (MQClientException e) {
            log.error("consumer is start !!! consumerGroup:{},namesrvAddr:{}", groupName, namesrvAddr, e);
            throw new RocketMQException(e);
        }
        return consumer;
    }
}
